package org.example.com.atguigu.day02;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.junit.Test;

import java.io.Serializable;
import java.util.Arrays;

public class $02_Transformation implements Serializable {

    /**
     * RDD的算子分为两种:
     *      Transformation算子[转换算子]： 会生成新的RDD, 不会触发任务的计算
     *      Action算子[行动算子]: 不会生成新RDD, 会触发任务的计算
     */


    /**
     * map: 一对一转换 【 原RDD一个元素通过计算之后会得到新RDD一个元素 】
     *      map需要传入一个函数,函数的参数只有一个,参数类型是原RDD元素类型,返回值类型无所谓
     *      map主要用于一对一的转换
     *      map类似sql的select
     *      map生成新RDD元素个数 = 原RDD元素个数
     */
    @Test
    public void map(){

        //1、创建JavaSparkcontext
        SparkConf conf = new SparkConf().setMaster("local[3]").setAppName("com.atguigu.day01.$01_RddCreate");
        JavaSparkContext sc = new JavaSparkContext(conf);

        JavaRDD<Integer> rdd1 = sc.parallelize(Arrays.asList(1, 6, 8, 9, 10, 2));

        //map中函数的第一个泛型是待计算的参数类型,第二个泛型是返回值类型

        JavaRDD<Integer> rdd2 = rdd1.map(new Function<Integer, Integer>() {
            @Override
            public Integer call(Integer v1) throws Exception {
                System.out.println(Thread.currentThread().getName() + "---" + v1);
                return v1 * 10;
            }
        });

        System.out.println(rdd2.collect());
    }
}

